Consumers start at StartLocation::Earliest, which will read historical events from the partition and can satisfy expected_events without ever consuming events from this test run. For a bounded publish/read validation, start from Latest (or from an enqueued time captured just before producers start) so the consumed set corresponds to the events published by this run.

expected_events is set to config.event_count for every consumer task. With multiple consumers, that makes the aggregate expected count event_count * consumer_count and can lead to unnecessary timeouts/hanging behavior (especially since consumers may share partitions). Consider dividing expected events per consumer based on partition assignment, or coordinating completion based on a shared consumed counter/map.

tests/stress_tests.rs defines its own main and is intended to be a custom-harness test binary, but there is no [[test]] entry in this crate’s Cargo.toml with harness = false for the stress_tests target. As-is, this integration test will fail to compile due to the default libtest harness generating its own main. Add a [[test]] name = "stress_tests" section with harness = false (and keep the file as the entrypoint).

Even with a harness = false test target, cargo test will execute this binary by default, but it requires live env vars and (by default) runs a 72h scenario. This is very likely to break local cargo test runs and CI. Consider gating the test target behind required-features (e.g., required-features = ["stress-tests"]) and adding a non-default feature, or moving this CLI into examples/ or a separate crate/binary so it never runs as part of normal tests.

The consumer selects a partition with consumer_id % 2, assuming exactly two partitions. This will fail for Event Hubs with a different partition count (invalid partition id) and will never read from partitions beyond 0/1. Query the hub’s partition_ids (as the continuous test does) and assign consumers across the real set of partitions.

This change adds clap/dotenvy dev-dependencies for the new stress-test CLI, but the crate is still missing a [[test]] section to declare the stress_tests integration test as harness = false (and ideally required-features = [...] to prevent it from running by default). Without that, the new tests/stress_tests.rs (which defines main) won’t compile as an integration test, and/or will run unexpectedly in CI.

TestConfig::default() values (event_count=1000, producer/consumer_count=2, etc.) don’t match the clap defaults / run_standalone fallbacks (100 events, 1 producer, 1 consumer). This makes the “default config” ambiguous and can lead to confusing behavior depending on whether callers use TestConfig::default() vs CLI. Consider aligning these defaults (either update Default or the clap .default_value(...) + unwrap_or(...) values).

producer_count and batch_size are taken from CLI without validation. producer_count == 0 will panic (division by zero in start_producers), and batch_size == 0 will cause an infinite loop in run_producer_task (batch_end never advances). Add clap validation (e.g., value_parser range >=1) and/or explicit checks that return a user-friendly error before running.

This test uses std::sync::Mutex for published_events / consumed_events and locks it inside async tasks. In a Tokio runtime this can block the executor thread and distort stress test results (and in worst cases lead to starvation). Prefer tokio::sync::Mutex (or a concurrent map like DashMap) for data shared across async tasks.

consumer_count is accepted via CLI and stored in TestConfig, but the implementation spawns one consumer per partition and only uses consumer_count for logging. This makes the CLI option misleading. Either honor consumer_count (e.g., spawn that many consumers and assign partitions) or remove the option/config field to avoid confusion.

When rolling over to a new batch, the result of batch.try_add_event_data(event, None)? is ignored. If an individual event can’t fit into an empty batch, try_add_event_data returns false and the event is silently not sent, but it is still inserted into missing_events and will be reported as lost. Handle the false case explicitly (return error / increment failure metric) before recording it as expected.

send_loop is spawned before receivers are opened, but receivers default to StartLocation::Latest. Any events sent before a given partition receiver attaches will never be observed, yet they’ve already been added to missing_events and will be counted as lost/corrupted. Start receivers before the sender and/or use an EnqueuedTime start position captured at test start so the receiver includes all events produced by this run.

missing_events tracks every sent event body until it’s received, and entries are never aged out. With a 72h default duration (and if receivers fall behind or errors occur), this map can grow without bound and cause excessive memory usage/OOM. Consider bounding it (e.g., per-partition watermarks, sampling, periodic pruning) or switching to metrics that don’t require storing every outstanding event key.

When no subcommand is provided, main runs all stress tests with default args. With continuous_send_receive_stress defaulting to 72 hours, cargo test --test stress_tests can unexpectedly run for days. Consider requiring an explicit subcommand (or using a much shorter default duration when invoked via the “run all” path) to avoid accidental long-running executions.

If try_add_event_data(...) returns false on an empty batch (which happens when a single event exceeds the max batch size), this code goes into the “batch is full” path and calls send_batch(batch, ...) on an empty EventDataBatch. EventDataBatch::get_messages() expects a batch envelope and will panic for empty batches. Detect the “event too large for an empty batch” case and return an error instead of trying to send the empty batch.

min_batch_size can be set to 0 via CLI (u32 parser allows it). That can lead to batch_size == 0, and the loop will attempt to send_batch with an empty EventDataBatch (panic) and will also skew metrics. Consider enforcing min-batch >= 1/max-batch >= 1 at the clap layer (or return an error when batch_size resolves to 0).